14 多Agent

前面的文章中，我们一直在和单个Agent打交道。但当你的应用越来越复杂，单个Agent可能会遇到这些问题：

工具太多，Agent选不好用哪个
不同任务需要不同的专业知识和提示词
有些任务可以并行处理，提高效率
不同团队想各自开发自己的功能

这时候就需要多Agent协作了——把一个大任务拆分给多个专业Agent，各司其职。

不过要先说一句：不是所有复杂任务都需要多Agent。很多时候，一个Agent配上合适的工具和提示词就够了。多Agent会增加复杂度和调用成本，只在确实需要的时候才用。

一、四种协作模式

LangChain提供了四种多Agent协作模式：

模式	核心思路	适合场景
子Agent	主Agent把子Agent当工具调用	多领域任务、需要集中控制
交接	通过状态切换动态改变Agent行为	客服、多阶段对话
路由	先分类再分发到专业Agent	明确的领域划分
自定义工作流	用LangGraph自己编排	复杂的业务流程

二、子Agent模式

这是最常用的模式。一个主Agent（Supervisor）负责协调，把具体的子任务分发给子Agent处理。子Agent就像员工，主Agent就像经理。

mermaid

graph LR
    A[用户] --> B[主Agent]
    B --> C[子Agent A]
    B --> D[子Agent B]
    B --> E[子Agent C]
    C --> B
    D --> B
    E --> B
    B --> F[回复用户]

2.1 基本实现

把子Agent包装成工具，主Agent通过调用工具来委托任务：

python

from langchain.tools import tool
from langchain.agents import create_agent

# 创建子Agent
research_agent = create_agent(
    model="deepseek-v4-flash",
    tools=[search_web, read_document],
    system_prompt="你是一个研究专家，擅长搜索和分析信息。",
)

writer_agent = create_agent(
    model="deepseek-v4-flash",
    tools=[],
    system_prompt="你是一个写作专家，擅长把信息组织成清晰的文章。",
)

# 把子Agent包装成工具
@tool
def research(query: str) -> str:
    """研究和搜索信息。当你需要查找资料、分析数据时使用。"""
    result = research_agent.invoke({
        "messages": [{"role": "user", "content": query}]
    })
    return result["messages"][-1].content

@tool
def write_content(topic: str, research_result: str) -> str:
    """根据研究结果撰写内容。当你需要把信息写成文章时使用。"""
    result = writer_agent.invoke({
        "messages": [{"role": "user", "content": f"主题：{topic}\n研究资料：{research_result}"}]
    })
    return result["messages"][-1].content

# 主Agent负责协调
main_agent = create_agent(
    model="deepseek-v4-flash",
    tools=[research, write_content],
    system_prompt=(
        "你是一个项目经理，负责协调研究和写作任务。"
        "- 需要查找资料时，使用research工具\n"
        "- 需要写文章时，使用write_content工具\n"
        "- 可以先研究再写作，也可以并行处理多个任务"
    ),
)

# 使用
result = main_agent.invoke({
    "messages": [{"role": "user", "content": "帮我写一篇关于AI发展趋势的简报"}]
})
print(result["messages"][-1].content)

2.2 子Agent的特点

集中控制：所有路由都经过主Agent，主Agent决定什么时候调用哪个子Agent
子Agent无状态：每次调用都是全新的，不记得之前的对话，所有记忆由主Agent维护
上下文隔离：每个子Agent在干净的上下文中工作，不会被主对话的历史消息干扰
可并行：主Agent可以同时调用多个子Agent

2.3 同步vs异步

默认情况下，子Agent调用是同步的——主Agent会等子Agent完成再继续。如果子Agent的任务耗时很长，可以用异步方式：

python

# 异步模式：三个工具配合使用
@tool
def start_background_task(task_description: str) -> str:
    """启动后台任务，返回任务ID"""
    job_id = start_job(task_description)
    return f"任务已启动，ID: {job_id}"

@tool
def check_task_status(job_id: str) -> str:
    """查询后台任务状态"""
    return get_job_status(job_id)

@tool
def get_task_result(job_id: str) -> str:
    """获取已完成任务的结果"""
    return get_job_result(job_id)

三、交接模式

交接模式的核心思想是：通过状态切换来改变Agent的行为。工具调用会更新一个状态变量（比如current_step），系统根据这个变量调整Agent的提示词和可用工具。

打个比方：就像一个客服人员，接到投诉电话后，先问保修信息（第一步），确认保修状态后切换到处理模式（第二步），整个过程中始终是同一个人在接电话，但他的行为在不同阶段是不同的。

3.1 用中间件实现

最简单的方式是用一个Agent配合中间件，根据状态动态切换配置：

python

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain.tools import tool, ToolRuntime
from langchain.messages import ToolMessage
from langgraph.types import Command
from typing import Callable


# 1. 定义带状态的State
class SupportState(AgentState):
    current_step: str = "triage"  # 当前步骤
    warranty_status: str | None = None


# 2. 工具通过Command更新状态
@tool
def record_warranty_status(
    status: str,
    runtime: ToolRuntime[None, SupportState],
) -> Command:
    """记录保修状态，切换到下一步"""
    return Command(update={
        "messages": [
            ToolMessage(
                content=f"保修状态已记录: {status}",
                tool_call_id=runtime.tool_call_id,
            )
        ],
        "warranty_status": status,
        "current_step": "specialist",  # 切换到专家模式
    })


@tool
def provide_solution(solution: str) -> str:
    """提供解决方案"""
    return f"解决方案: {solution}"


# 3. 中间件根据状态动态切换配置
@wrap_model_call
def apply_step_config(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    step = request.state.get("current_step", "triage")

    configs = {
        "triage": {
            "prompt": "你是客服分诊员。收集用户的保修信息，使用record_warranty_status记录。",
            "tools": [record_warranty_status],
        },
        "specialist": {
            "prompt": "你是售后专家。根据保修状态提供解决方案。",
            "tools": [provide_solution],
        },
    }

    config = configs[step]
    modified_request = request.override(
        system_prompt=config["prompt"],
        tools=config["tools"],
    )
    return handler(modified_request)


# 4. 创建Agent
agent = create_agent(
    model="deepseek-v4-flash",
    tools=[record_warranty_status, provide_solution],
    state_schema=SupportState,
    middleware=[apply_step_config],
)

# 用户对话会自动在不同阶段切换
result = agent.invoke({
    "messages": [{"role": "user", "content": "我的手机屏幕碎了，还在保修期内"}]
})

3.2 适用场景

客服系统：分诊 → 确认信息 → 解决问题
多阶段对话：收集信息 → 分析 → 给出建议
需要顺序约束的场景：必须先完成A才能做B

四、路由模式

路由模式是先对用户输入做分类，然后分发给对应的专业Agent处理。就像医院的导诊台，先判断你是什么病，再让你去对应的科室。

mermaid

graph LR
    A[用户输入] --> B[路由器]
    B --> C[Agent A]
    B --> D[Agent B]
    B --> E[Agent C]
    C --> F[汇总结果]
    D --> F
    E --> F
    F --> G[回复用户]

4.1 单Agent路由

用Command把请求路由到一个专业Agent：

python

from langgraph.types import Command
from langchain.agents import create_agent

# 创建专业Agent
sales_agent = create_agent(
    model="deepseek-v4-flash",
    tools=[check_price, place_order],
    system_prompt="你是销售顾问，帮助用户了解产品和下单。",
)

support_agent = create_agent(
    model="deepseek-v4-flash",
    tools=[check_order, file_ticket],
    system_prompt="你是客服代表，帮助用户解决售后问题。",
)

# 路由函数：分类用户输入
def classify_and_route(state) -> Command:
    last_message = state["messages"][-1].content

    # 简单的关键词分类，实际项目可以用LLM分类
    if any(word in last_message for word in ["价格", "购买", "下单", "产品"]):
        return Command(goto="sales_agent")
    else:
        return Command(goto="support_agent")

4.2 并行路由

如果一个问题需要多个领域同时查询，可以用Send并行分发：

python

from langgraph.types import Send

def route_to_agents(state):
    """把查询分发给多个Agent并行处理"""
    query = state["query"]

    return [
        Send("github_agent", {"query": query}),
        Send("notion_agent", {"query": query}),
        Send("slack_agent", {"query": query}),
    ]

4.3 有状态路由

无状态路由每次都要重新分类。如果需要多轮对话，可以把路由包装成工具，让一个对话Agent来调用：

python

@tool
def search_docs(query: str) -> str:
    """搜索多个文档源"""
    result = router_workflow.invoke({"query": query})
    return result["final_answer"]

# 对话Agent使用路由作为工具
agent = create_agent(
    model="deepseek-v4-flash",
    tools=[search_docs],
    system_prompt="你是一个助手，使用search_docs工具回答问题。",
)

五、自定义工作流

当上面的模式都不能满足需求时，可以用LangGraph自己编排执行流程。你有完全的控制权：顺序执行、条件分支、循环、并行，想怎么编排就怎么编排。

5.1 基本结构

python

from langchain.agents import create_agent
from langgraph.graph import StateGraph, START, END

# 创建Agent
agent = create_agent(model="deepseek-v4-flash", tools=[...])

# 定义节点：在LangGraph节点中调用Agent
def agent_node(state: dict) -> dict:
    result = agent.invoke({
        "messages": [{"role": "user", "content": state["query"]}]
    })
    return {"answer": result["messages"][-1].content}

# 构建工作流
workflow = (
    StateGraph(State)
    .add_node("agent", agent_node)
    .add_edge(START, "agent")
    .add_edge("agent", END)
    .compile()
)

5.2 实际例子：RAG流水线

一个典型的RAG工作流包含三个阶段：查询改写 → 文档检索 → Agent生成答案：

python

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END


# 定义状态
class State(TypedDict):
    question: str
    rewritten_query: str
    documents: list[str]
    answer: str


# 节点1：用LLM改写查询，提高检索质量
def rewrite_query(state: State) -> dict:
    model = ChatOpenAI(model="deepseek-v4-flash")
    response = model.invoke([
        {"role": "system", "content": "把用户的问题改写成更适合搜索的形式。只返回改写后的查询。"},
        {"role": "user", "content": state["question"]},
    ])
    return {"rewritten_query": response.content}


# 节点2：检索文档（纯确定性逻辑，不需要LLM）
def retrieve(state: State) -> dict:
    docs = vector_store.similarity_search(state["rewritten_query"], k=5)
    return {"documents": [doc.page_content for doc in docs]}


# 节点3：Agent根据检索结果生成答案
def generate_answer(state: State) -> dict:
    agent = create_agent(model="deepseek-v4-flash", tools=[...])
    context = "\n\n".join(state["documents"])
    result = agent.invoke({
        "messages": [{"role": "user", "content": f"参考资料：\n{context}\n\n问题：{state['question']}"}]
    })
    return {"answer": result["messages"][-1].content}


# 构建工作流
workflow = (
    StateGraph(State)
    .add_node("rewrite", rewrite_query)
    .add_node("retrieve", retrieve)
    .add_node("agent", generate_answer)
    .add_edge(START, "rewrite")
    .add_edge("rewrite", "retrieve")
    .add_edge("retrieve", "agent")
    .add_edge("agent", END)
    .compile()
)

result = workflow.invoke({"question": "2024年WNBA总冠军是谁？"})
print(result["answer"])

六、怎么选？

场景	推荐模式
多领域任务，需要集中控制	子Agent
多阶段对话，需要状态切换	交接
输入明确分为几个类别	路由
需要复杂的执行流程	自定义工作流
只有几个工具	单Agent就够了

几个判断依据：

子Agent：适合需要并行处理、分布式开发的场景，主Agent保持对全局的掌控
交接：适合需要和用户直接对话、按步骤推进的场景
路由：适合输入可以明确分类的场景，实现简单
自定义工作流：适合标准模式都不够用的场景，灵活但复杂

这些模式也可以组合使用。比如子Agent内部可以用路由模式，路由Agent可以调用自定义工作流。

七、总结

多Agent协作让复杂任务变得可管理：

子Agent模式：主Agent把子Agent当工具调用，集中控制，可并行
交接模式：通过状态切换动态改变Agent行为，适合多阶段对话
路由模式：先分类再分发，适合明确的领域划分
自定义工作流：用LangGraph自由编排，最灵活也最复杂

选择哪种模式取决于你的具体需求。记住：能用单Agent解决的问题，就不要用多Agent。多Agent是为了应对真正的复杂性，而不是为了看起来酷。

在下一篇文章中，我们将学习人在环路（Human-in-the-Loop），看看如何让人类参与到Agent的决策过程中。

14 多Agent ​

一、四种协作模式 ​

二、子Agent模式 ​

2.1 基本实现 ​

2.2 子Agent的特点 ​

2.3 同步vs异步 ​

三、交接模式 ​

3.1 用中间件实现 ​

3.2 适用场景 ​

四、路由模式 ​

4.1 单Agent路由 ​

4.2 并行路由 ​

4.3 有状态路由 ​

五、自定义工作流 ​

5.1 基本结构 ​

5.2 实际例子：RAG流水线 ​

六、怎么选？ ​

七、总结 ​

14 多Agent

一、四种协作模式

二、子Agent模式

2.1 基本实现

2.2 子Agent的特点

2.3 同步vs异步

三、交接模式

3.1 用中间件实现

3.2 适用场景

四、路由模式

4.1 单Agent路由

4.2 并行路由

4.3 有状态路由

五、自定义工作流

5.1 基本结构

5.2 实际例子：RAG流水线

六、怎么选？

七、总结